package com.atguigu.flink.sql.function;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/4/14
 *
 *  可以使用TableAPI，可以使用SQL。
 *  可以使用DataStreamAPI。
 *      DataStreamSource  ===== Table  ======  表名
 *
 */
public class Demo1_StreamTableConvert
{
    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        Schema schema = Schema.newBuilder()
                              .column("id", "STRING")
                              .column("ts", "BIGINT")
                              .column("vc", "INT")
                              .columnByExpression("pt", "proctime()")
                              .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                              .build();
        //把流转换为 Table
        Table table = tableEnvironment.fromDataStream(ds,schema);

        //把Table转换为表名
        tableEnvironment.createTemporaryView("t1",table);

        //使用sql查询
        Table t2 = tableEnvironment.sqlQuery("select * from t1 where id > 's1' ");

        /*
            使用TableAPI操作

            +I: 第一次向表中插入一行记录

            聚合操作:
            -U： 一条记录被更新了，更新前的样子
            +U： 一条记录被更新了，更新后的样子
         */
        Table t3 = t2
            .groupBy($("id"))
            .select($("id"), $("vc").sum().as("sumVc"));

        t3.printSchema();

        /*
            把 Table 转换为 DataStreamSource
                toChangelogStream() : 适用于聚合的场景。 数据会随着聚合随时更新。
                                        也适用于非聚合。
                                            流中有 insert，update,delete
                                            insert: +I
                                            update: -U,+U
                                            delete:  -D

                toDataStream() : 适用于非聚合的场景，运算是没有聚合。
                                返回的是一个 Append-only 流。 流中只有insert +I。

                 Row:  flink自带的类型，用于封装一行(很多列)数据。
         */
        DataStream<Row> ds2 = tableEnvironment.toChangelogStream(t3);


        ds2
            .map(new MapFunction<Row, String>()
            {
                @Override
                public String map(Row value) throws Exception {
                    String id = value.<String>getFieldAs("id");
                    Integer sumVc = value.<Integer>getFieldAs("sumVc");
                    return id + ":" +sumVc;
                }
            })
            .print();


        env.execute();

    }
}
